
package peerToPeerFacebook.proj;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;

import peerToPeerFacebook.src.edu.washington.cs.cse490h.lib.Utility;

/**
 * This calls implements the role of proposer in Paxos implementation.
 */
public class PaxosProposer
{
    /** The FBNode reference */
    FBNode node;

    /** The proposal Number used for proposing a new value */
    private Integer proposalNo;

    /** The file which stores the proposal Number */
    private String proposalNoFile = "proposalNoFile.txt";

    /** The list of all the servers participating in the paxos algorithm */
    List<Integer> serverList = new ArrayList<Integer>();

    /** The name of the prepare Method **/
    String prepareMethodName = "prepare";

    /** The name of the accept Method **/
    String acceptMethodName = "accept";

    /** The name of the learn Method **/
    String learnMethodName = "learn";

    /** The instance no of the current paxos instance **/
    int instanceNo;

    /** The no of acceptor who promised **/
    int noOfPromises;

    /**
     * The integer count which represents the how many servers failed to promise
     * the prepare Request
     **/
    int noOfNo;

    /** the integer which represents the no of servers accepted the proposal **/
    int noOfProposalAccepted;

    /** The integer which represents the total no of Accept messages sent **/
    int totalAcceptSent;

    /**
     * The integer count which represents the how many servers responded to
     * prepare requests
     **/
    int count;

    /** The delimiter used for proposal Nos **/
    String proposalDelimiter = ":";

    /** The list of the responses received **/
    ArrayList<PrepareResponse> responseList;

    /** The paxos command proposed for this round of paxos **/
    PaxosCommand command;

   /**
     * This method informs ONLY local PSM about the outcome of the round. 
     * In this case some value was already chosen and the local PSM may have missed
     * that value so by informing local PSM we can ensure that we knows about all the missed
     * values. 
     * 
     * @param chosenValue, the value which has already been chosen
     * @param someValueIsChosen, the boolean flag which indicates if some value
     *            was already chosen or not.
     */
    private void informLocalPSM(PaxosCommand chosenValue, boolean someValueIsChosen)
    {
        //System.out.println(" \n\n\n Going to inform local PSM about the value chosen");
        RPCClient rpcClient = node.getRPCClient(this.node.addr);
        if (someValueIsChosen)
            rpcClient.invoke(learnMethodName, new Serializable[] {this.instanceNo,
                    chosenValue, false}, new IResultCallback()
            {
                public void resultCallback(Status status, Serializable result)
                {
                    // empty
                }
            });
        else
            rpcClient.invoke(learnMethodName,
                             new Serializable[] {this.instanceNo, null, false},
                             new IResultCallback()
                             {
                                 public void resultCallback(Status status, Serializable result)
                                 {
                                     // empty
                                 }
                             });
    }

    /**
     * Helper method which re-initializes the class member variables
     */
    private void clearVariables()
    {
        this.noOfPromises = 0;
        this.noOfNo = 0;
        this.responseList = new ArrayList<PrepareResponse>();
        this.count = 0;
    }

    /**
     * Constructor of the Paxos Proposer
     * 
     * @param node, the FBNode reference
     * @param serverList, the list of servers participating in the paxos group
     */
    public PaxosProposer(FBNode node, List<Integer> serverList)
    {
        this.node = node;
        this.serverList = serverList;
        clearVariables();
    }

    /**
     * This method is invoked by the PSM when he wants to propose a new value.
     * This method ensures to use the proper proposal No and then calls sendPrepare
     * whicn sends prepare method to all the proposers.
     * 
     * @param command, the command that needs to be proposed.
     * @param instanceNo, the paxos instance for which the command will be
     *            proposed.
     */
    public void propose(PaxosCommand command, int instanceNo)
    {
        this.command = command;
        this.instanceNo = instanceNo;
        if (!Utility.fileExists(node, proposalNoFile))
        {
            this.proposalNo = 1;
            UtilityFile.writeToFile(node, proposalNoFile, proposalNo.toString());
        }
        else
        {
            // read the last proposal No used
            proposalNo = Integer.parseInt(UtilityFile.readFromFile(this.node, proposalNoFile));
            assert proposalNo >= 0 : "The proposal No should be greater than zero.";
            // Increment the proposal No and write it back to the file
            this.proposalNo++;
            UtilityFile.writeToFile(node, proposalNoFile, proposalNo.toString());
        }
        sendPrepareMessage();
    }

    /**
     * This method sends prepare message to all the servers in the paxos group
     * and after receiving responses it evalutes to responses of the acceptors.
     */
    private void sendPrepareMessage()
    {
        RPCClient rpcClient;
        Iterator<Integer> itr = serverList.iterator();

        while (itr.hasNext())
        {
            final int serverID = itr.next();
            rpcClient = node.getRPCClient(serverID);
            String proposal =
                    this.proposalNo.toString() + this.proposalDelimiter + this.node.addr;
            System.out.println("\n\n PP: PaxosInstance: " + this.instanceNo +
                               " Calling prepare with proposal No : " + proposal);
            rpcClient.invoke(prepareMethodName, new Serializable[] {
                                     this.instanceNo,
                                     this.proposalNo.toString() + proposalDelimiter +
                                             node.addr},
                             new IResultCallback()
                             {
                                 @Override
                                 public void resultCallback(Status status, Serializable result)
                                 {
                                     count++;
                                     if (status == Status.Success)
                                     {
                                         // Call succeeded, lets evaluate the response
                                         PrepareResponse response = (PrepareResponse) result;
                                         assert response != null : "The prepare Response should not be null";
                                         evaluateResponse(response);
                                     }
                                     else
                                     {
                                         System.out
                                                 .println(" Server: " + serverID +
                                                          " failed to respond to the Prepare Request message");
                                         evaluateResponse(null);

                                     }
                                 }
                             });

        }
    }

    /**
     * This method that evaluates the response received from the acceptor
     * to check if majority agreed to accept the proposal in which case he tries to pass the
     * second round (accept phase) of the algorithm. Otherwise it calls other methods which evaluate
     * the responses received and start new round of paxos Or inform local PSM depending on the responses.
     * 
     * @param response, the response received from the acceptor
     */
    private void evaluateResponse(PrepareResponse response)
    {
        if(response != null)
           //System.out.println(" \n\n\n ++++++++++ The response recieved is (proposal No): " + response.proposalNo +" (accepted value);  " + response.AcceptedValue + " (promise received); " + response.promise + " for instance No: " + this.instanceNo + "+++++++ \n\n\n"); 
        if (response != null)
        {
            if (response.promise)
            {
                this.noOfPromises++;
                this.responseList.add(response);
            }
            else
            {
                this.noOfNo++;
                this.responseList.add(response);
            }
        }

        // Needs to check if all replies have been received, its time to process
        // responses and decide what should be done next 
        if (this.count == this.serverList.size())
        {
            // check all responses to see which value should be proposed
            if (this.noOfPromises > (count / 2))
            {
                System.out
                        .println("\n\n Majority responded to the prepare request... So now calculating the value which needs to be proposed\n\n");
                System.out.println("\n\n Total promise received: " + this.noOfPromises +
                                   "\n\n");
                // Majority replied with promise, So proposer can now send
                // accept message.
                // Let check which value should be used for accept message
                calculateToBeProposedValue(this.responseList);
            }
            else if (this.noOfNo > (count / 2))
            {
                System.out
                        .println("\n\n Majority didn't return the promise... lets evaluate the responses. \n\n");
                System.out.println("\n\n Total No received: " + this.noOfNo + "\n\n");
                // Majority said No, so accept message cannot be sent. Lets
                // check if some value was has already been accepted
                // In that case we need to inform the caller about the learned
                // value and if not then we need to start another round.
                InformPSMAboutOutcome(this.responseList);
            }
            else
            {
                // In all other cases we should start a new paxos round.
                System.out
                        .println(" \n\n Neither majority said Yes nor No....So starting a new round of Paxos\n\n");
                startNewPaxosRound(responseList);
            }
        }
    }

    /**
     * This method checks the response list to see what value need to be sent for 
     * accept phase. If acceptors responded with some last accepted values proposer uses 
     * that value in the accept phase otherwise it uses the "proposed command" (Which it got from the local PSM in propose request)
     * to send the accept message.
     * 
     * @param responseList, the response List
     */
    private void calculateToBeProposedValue(ArrayList<PrepareResponse> responseList)
    {
        ProposalComparator comparator = new ProposalComparator();
        PrepareResponse highestAccepted = null;

        // Lets check is we have got some last accepted value from any acceptor.
        for (PrepareResponse response : responseList)
            if (response.AcceptedValue != null)
                if (highestAccepted == null ||
                    (highestAccepted != null && comparator.compare(response.proposalNo,
                                                                   highestAccepted.proposalNo) > 0))
                    highestAccepted = response;

        if (highestAccepted != null)
        {
            // Now we have value to be proposed, we can call accept method with
            // this value. This will not propose the value the client wants to
            // propose. This round of accept will just make sure
            // that all paxos group servers accept the last accepted value and are in sync.
            //System.out.println("\n\n\n Going to send highest Accepted " +  highestAccepted.AcceptedValue +" value for accept phase \n\n\n");
            sendAccept(this.proposalNo.toString() + proposalDelimiter + node.addr,
                       highestAccepted.AcceptedValue);
        }
        else
        {
            // There was no accepted value returned by the servers, so now the
            // value client wants to propose can be sent in accept round.
            //System.out.println("\n\n\n Going to send the PROPOSED value for accept phase \n\n\n");
            sendAccept(this.proposalNo.toString() + proposalDelimiter + node.addr,
                       this.command);
        }
    }

    /**
     * This method checks to see if some value has already been accepted in 
     * previous rounds in which case it informs the PSM, otherwise it starts a new round
     * 
     * @param responseList, the responseList received from Acceptors
     */
    private void InformPSMAboutOutcome(ArrayList<PrepareResponse> responseList)
    {
        PaxosCommand chosenValue = null;
        boolean valueWasChosen = false;
     
        Iterator<PrepareResponse> itr1 = responseList.iterator();
        while (itr1.hasNext())
        {
            int equalElementCount = 0;
            PrepareResponse first = itr1.next();
            Iterator<PrepareResponse> itr2 = responseList.iterator();
            while (itr2.hasNext())
            {
                PrepareResponse second = itr2.next();
                if (first.proposalNo != null && second.proposalNo != null &&
                    first.proposalNo.equals(second.proposalNo) &&
                    !(first.AcceptedValue == null) && !(second.AcceptedValue == null) &&
                    first.AcceptedValue.equals(second.AcceptedValue))
                {
                    chosenValue = first.AcceptedValue;
                    equalElementCount++;
                }
                // found majority of elements which are equal, it means some
                // value was accepted, Needs to inform local PSM.
                if (equalElementCount > (count / 2))
                {
                    valueWasChosen = true;
                    break;
                }
            }
        }
        if (valueWasChosen) // Inform local PSM that some value is already
            // learned for that particular instance.
            informLocalPSM(chosenValue, true);
        else
            // No value was chosen and majority didn't return a promise. Need to
            // start a new round with higher proposal No
            startNewPaxosRound(responseList);
    }

    /**
     * This method sends the accept messages to all the acceptors in the paxos
     * group and after all messages are sent it checks to make sure that all acceptors accepted
     * the messages in which case it informs all PSMs about the chosen value otherwise it starts a new round again.
     * 
     * @param proposalNo, the proposal No of the accept message
     * @param valueToBeProposed, the value which needs to be accepted.
     */
    private void sendAccept(final String proposalNo, final PaxosCommand valueToBeProposed)
    {
        RPCClient rpcClient;
        Iterator<Integer> itr = serverList.iterator();
        noOfProposalAccepted = 0;
        totalAcceptSent = 0;

        while (itr.hasNext())
        {
            final int serverID = itr.next();
            rpcClient = node.getRPCClient(serverID);
            rpcClient.invoke(acceptMethodName, new Serializable[] {this.instanceNo,
                    proposalNo, valueToBeProposed}, new IResultCallback()
            {
                @Override
                public void resultCallback(Status status, Serializable result)
                {
                    if ((status == Status.Success) && ((boolean) result.equals(true)))
                    {
                        // if majority accepted, then we can inform PSMs
                        noOfProposalAccepted++;
                    }
                    else
                    {
                        // this is just a debug print
                        System.out.println(" Server: " + serverID +
                                           " failed to accept the proposal Request: " + proposalNo);
                    }
                    totalAcceptSent++;
                    if (totalAcceptSent == serverList.size())
                    {
                        if (noOfProposalAccepted > (totalAcceptSent / 2))
                        {
                            // Value got accepted so lets inform the PSM about
                            // the success
                            informAllPSMs(valueToBeProposed, true);
                        }
                        else
                        {
                            // may be the majority is down Or the acceptors have responded 
                            // to higher numbered proposal, so lets start another round of paxos.
                            startNewPaxosRound(null);
                        }
                    }
                }
            });
        }
    }

    /**
     * A helper method that informs all PSMs about the success of the value
     * accepted by majority of paxos group
     * 
     * @param valueChosen, the value which has been accepted by majority
     * @param valueAccepted, the boolean flag which indicates whether the value
     *            proposed by the client has been accepted or some other value
     */
    private void informAllPSMs(PaxosCommand valueChosen, boolean valueAccepted)
    {
        //System.out.println(" \n\n\n Going to inform everyone about the value chosen \n\n\n");
        assert valueChosen != null : "Node should never accept null value";

        Iterator<Integer> itr = serverList.iterator();
        while (itr.hasNext())
        {
            final int serverID = (int) itr.next();
            RPCClient rpcClient = node.getRPCClient(serverID);
            // If its the local PSM, we need to inform that "his" value got
            // accepted, for others we need to inform that the "value accepted"
            // was proposed by some other node
            if (serverID == this.node.addr)
            {
                rpcClient.invoke(learnMethodName,
                                 new Serializable[] {this.instanceNo, valueChosen,
                                         valueAccepted && valueChosen.equals(this.command)},
                                 new IResultCallback()
                                 {
                                     public void resultCallback(Status status,
                                                                Serializable result)
                                     {
                                         // empty
                                     }
                                 });
            }
            else
            {
                rpcClient.invoke(learnMethodName, new Serializable[] {this.instanceNo,
                        valueChosen, false}, new IResultCallback()
                {
                    public void resultCallback(Status status, Serializable result)
                    {
                        // empty
                    }
                });
            }
        }
    }

    /**
     * This method starts a new round of Paxos with the highest proposal No
     * received in the previous Round, if any, otherwise increments the local
     * proposal No and uses that.
     * 
     * @param responseList, the response list from the previous round
     */
    void startNewPaxosRound(ArrayList<PrepareResponse> responseList)
    {
        System.out.println(" \n\n Going to start a new round of paxos \n\n");
        int newProposalNo;
        if (responseList != null)
        {
            ProposalComparator comparator = new ProposalComparator();
            PrepareResponse highestProposalResponse = null;
            for (PrepareResponse response : responseList)
                if (response.proposalNo != null)
                    if (highestProposalResponse == null ||
                        (highestProposalResponse != null && comparator
                                .compare(response.proposalNo,
                                         highestProposalResponse.proposalNo) > 0))
                        highestProposalResponse = response;
            
           if (highestProposalResponse != null)
            {
                newProposalNo = getProposalNo(highestProposalResponse.proposalNo);
                // if the new proposal No is greater than the one we already have, lets use the greater one
                if(newProposalNo > proposalNo)
                {
                    //System.out.println(" \n\n\n Received proposal No is greater than the proposer's one. So going to use that. \n\n\n");
                    newProposalNo++;
                    this.proposalNo = newProposalNo;
                }
                else
                    this.proposalNo++;
            }
            else
                this.proposalNo++;
        }
        else
        {
            this.proposalNo++;
        }
        UtilityFile.writeToFile(node, proposalNoFile, proposalNo.toString());
        // Start a new round of paxos
        clearVariables();
        sendPrepareMessage();
    }

    /**
     * This method returns the int proposal No from the string proposal No which
     * contains both proposal No and node ID
     * 
     * @param proposalNo, the string proposal no (proposal No + node ID)
     * @return, the int proposal No
     */
    private int getProposalNo(String proposalNo)
    {
        String tokens[] = proposalNo.split(":");
        assert tokens.length == 2 : " The length of the proposal token should be equal to 2";
        return Integer.parseInt(tokens[0]);
    }
}

/**
 * A helper class that implements comparator which can be passed to
 * collections.sort function
 * 
 */
class ProposalComparator implements Comparator<String>
{
    /**
     * Method for comparing two string proposal No
     * 
     * @param one, the first proposalNo
     * @param two, the second proposal No
     * @return, 0 if both are equal, >1 or <1 depending if first is greater or
     *          less than the second one
     */
    @Override
    public int compare(String one, String two)
    {
        if (one.equals(null) || one.equals(""))
            return -1;
        if (two.equals(null) || two.equals(""))
            return 1;
        if ((one.equals(null) && two.equals(null)) || (one.equals("") && two.equals("")))
            return 0;
        String delims = "[:]+";
        String[] array1, array2;
        array1 = one.split(delims);
        array2 = two.split(delims);
        int round1 = Integer.parseInt(array1[0]);
        int round2 = Integer.parseInt(array2[0]);
        int result = 0, temp;

        if (round1 < round2)
        {
            result = -1;
        }
        else if (round1 > round2)
        {
            result = 1;
        }
        else if (round1 == round2)
        {
            temp = array1[1].compareTo(array2[1]);
            if (temp < 0)
            {
                result = -1;
            }
            else if (temp > 0)
            {
                result = 1;
            }
        }
        return result;
    }
}
